#include "head.h"
int main(int argc, char const *argv[])
{
    //打开文件,读取json文件，将josn文件保存在buf里
    FILE *file = fopen("node.json", "r");
    char buf[SIZE];
    int len = fread(buf, 1, SIZE, file);
    if (len < 0)
    {
        perror("fread err");
        return -1;
    }
    fclose(file);
    //把json字符串转换成一棵json树结构
    cJSON *root = cJSON_Parse(buf);
    //解析模式节点
    cJSON *item = cJSON_GetObjectItem(root, "report");
    cJSON *temp = cJSON_GetObjectItem(item, "type");
    type = temp->valueint; //模式
    temp = cJSON_GetObjectItem(item, "period");
    period = temp->valueint; //周期
    //解析mqtt服务器ip和端口号
    item = cJSON_GetObjectItem(root, "mqtt_server");
    temp = cJSON_GetObjectItem(item, "addr");
    ip = temp->valuestring; //ip
    temp = cJSON_GetObjectItem(item, "port");
    port = temp->valueint; //端口号
    //解析stm32节点
    item = cJSON_GetObjectItem(root, "stm32");
    cJSON *array = cJSON_GetObjectItem(item, "data");
    lenth = cJSON_GetArraySize(array);
    //解析modbus节点
    item = cJSON_GetObjectItem(root, "modbus");
    array = cJSON_GetObjectItem(item, "data");
    lenth1 = cJSON_GetArraySize(array);
    int sum = lenth + lenth1; //得到共享内存大小

    printf("%d\n",sum);
    //创建共享内存和消息队列
    int ret = -1;
    ret = shm_init(&para, "shm_test", MAX_NODE_SIZE);
    if (ret < 0)
    {
        return -1;
    }
    //node_p指针指向开辟共享内存空间
    node_p = shm_getaddr(&para);
    if (node_p == NULL)
    {
        return -1;
    }
    memset(node_p, 0, sizeof(struct std_node) * sum + 4);
    //前4个字节存储实际的设备数目
    total = (int *)node_p;
    //后面空间存储数据点
    node_arr = (struct std_node *)(node_p + sizeof(int));
    *total = sum; //设备数量
    //继续解析json格式，将32和modbus数组中的值写入共享内存
    item = cJSON_GetObjectItem(root, "stm32");
    array = cJSON_GetObjectItem(item, "data");
    lenth = cJSON_GetArraySize(array);
    //循环读取stm32数组中的数据放入共享内存,
    for (int i = 0; i < lenth; i++)
    {
        item = cJSON_GetArrayItem(array, i);//反序列化
        cJSON *data = cJSON_GetObjectItem(item, "key");
        node_arr[i].key = data->valueint;//拿到key值放入共享内存
        printf("%s=%d\n", data->string, data->valueint);
        printf("%d\n", node_arr[i].key);
        data = cJSON_GetObjectItem(item, "name");
        strcpy(node_arr[i].dev_type, data->valuestring);//拿到name放入共享内存
        printf("%s=%s\n", data->string, data->valuestring);
        printf("%s\n", node_arr[i].dev_type);
        data = cJSON_GetObjectItem(item, "type");
        node_arr[i].type = data->type;//拿到type值放入共享内存
        printf("%s=%d\n", data->string, data->valueint);
        printf("%d\n", node_arr[i].type);
        node_arr[i].old_val = node_arr[i].new_val; //吧当前值给旧值
        if (data->type == 1)
        {
            temp = cJSON_GetObjectItem(item, "data");
            node_arr[i].new_val.b_val = temp->valueint;
            printf("%s=%d\n", temp->string, temp->valueint);
            printf("%d\n", node_arr[i].new_val.b_val);
        }
        else if (data->type == 2)
        {
            temp = cJSON_GetObjectItem(item, "data");
            node_arr[i].new_val.i_val = temp->valueint;
            printf("%s=%d\n", temp->string, temp->valueint);
            printf("%d\n", node_arr[i].new_val.i_val);
        }
        else if (data->type == 3)
        {
            temp = cJSON_GetObjectItem(item, "data");
            node_arr[i].new_val.f_val = temp->valuedouble;
            printf("%s=%f\n", temp->string, temp->valuedouble);
            printf("%f\n", node_arr[i].new_val.f_val);
        }
        printf("----------------------------");
    }
    //继续解析json格式，将32和modbus数组中的值写入共享内存
    item = cJSON_GetObjectItem(root, "modbus");
    array = cJSON_GetObjectItem(item, "data");
    lenth1 = cJSON_GetArraySize(array);
    //循环读取modbus数组中的数据放入共享内存,
    for (int i = 0; i < lenth1; i++)
    {
        item = cJSON_GetArrayItem(array, i);
        cJSON *data = cJSON_GetObjectItem(item, "key");
        node_arr[i + lenth].key = data->valueint;
        printf("%s=%d\n", data->string, data->valueint);
        data = cJSON_GetObjectItem(item, "name");
        //node_arr[i + lenth].dev_type = &data->valuestring;
        strcpy(node_arr[i + lenth].dev_type, data->valuestring);
        printf("%s=%s\n", data->string, data->valuestring);
        data = cJSON_GetObjectItem(item, "type");
        node_arr[i + lenth].type = data->valueint;
        printf("%s=%d\n", data->string, data->valueint);
        if (data->type == 1)
        {
            temp = cJSON_GetObjectItem(item, "data");
            node_arr[i + lenth].new_val.b_val = temp->valueint;
            printf("%s=%d\n", temp->string, temp->valueint);
            printf("%d\n", node_arr[i + lenth].new_val.b_val);
        }
        else if (data->type == 2)
        {
            temp = cJSON_GetObjectItem(item, "data");
            node_arr[i + lenth].new_val.i_val = temp->valueint;
            printf("%s=%d\n", temp->string, temp->valueint);
            printf("%d\n", node_arr[i + lenth].new_val.i_val);
        }
        else if (data->type == 3)
        {
            temp = cJSON_GetObjectItem(item, "data");
            node_arr[i + lenth].new_val.f_val = temp->valuedouble;
            printf("%s=%f\n", temp->string, temp->valuedouble);
            printf("%f\n", node_arr[i + lenth].new_val.f_val);
        }
        printf("----------------------------");
    }
    //创建mqtt发送接受的两个线程
    pthread_t publish_tid, subscribe_tid, publish_h_tid, sqlite_tid;
    pthread_create(&publish_tid, NULL, publish, NULL);     //发送上报数据线程
    pthread_create(&subscribe_tid, NULL, subscribe, NULL); //接收数据线程
    pthread_create(&publish_h_tid, NULL, publish_h, NULL); //发送历史记录线程
    pthread_create(&sqlite_tid, NULL, sqlite, NULL);       //接收数据线程
    while (1)                                              //循环阻塞，防止进程结束
    {
    }

    return 0;
}
void delivered(void *context, MQTTClient_deliveryToken dt)
{
    //printf("Message with token value %d delivery confirmed\n", dt);
    deliveredtoken = dt;
}

//接收qt发来的消息解析
int msgarrvd(void *context, char *topicName, int topicLen, MQTTClient_message *message)
{
    printf("%s\n", topicName);
    if (strcmp(topicName,"data_recv") == 0)
    {
        int i;
        char *payloadptr;
        payloadptr = message->payload;
        //把json字符串转换成一棵json树结构
        cJSON *root = cJSON_Parse(payloadptr);
        //解析节点
        cJSON *item = cJSON_GetObjectItem(root, "name"); //拿到节点的name
        printf("----%d\n", __LINE__);
        strcpy(name_k, item->valuestring);
        printf("name = %s\n", name_k);

        item = cJSON_GetObjectItem(root, "key"); //拿到节点的key
        key_k = item->valueint;
        printf("key_k = %d\n", key_k);

        item = cJSON_GetObjectItem(root, "flag"); //拿到节点的flag
        flag_k = item->valueint;

        item = cJSON_GetObjectItem(root, "refresh"); //拿到刷新标志位
        refresh = item->valueint;

        printf("flag_k = %d\n", flag_k);
        //创建消息队列发送
        struct msgbuf send_buf;
        //这个mtype可以不用，但是必须赋一个不小于0的数
        //给消息队列结构体赋值
        send_buf.mtype = 1;
        strcpy(send_buf.name, name_k); //用来判断是给32还是modbus发
        send_buf.key = key_k;          //要发送的key值（对应那个设备）
        send_buf.flag = flag_k;        //要发送的通断标志位
        //将消息队列结构体发送
        if (strcmp(name_k, "stm32") == 0)
        {
            if (msg_queue_send("stm32", &send_buf, sizeof(send_buf), 0) < 0)
            {
                printf("msg_queue_send error\n");
                return -1;
            }
        }
        else if (strcmp(name_k, "modbus") == 0)
        {
            if (msg_queue_send("modbus", &send_buf, sizeof(send_buf), 0) < 0)
            {
                printf("msg_queue_send error\n");
                return -1;
            }
        }
        cJSON_Delete(root);
        putchar('\n');
        MQTTClient_freeMessage(&message);
        MQTTClient_free(topicName);
    }
    if (strcmp(topicName,"history_recv") == 0)
    {
        int i;
        char *payloadptr;
        payloadptr = message->payload;
        //把json字符串转换成一棵json树结构
        cJSON *root = cJSON_Parse(payloadptr);
        //解析节点
        cJSON *item = cJSON_GetObjectItem(root, "key"); //拿到节点的key_l
        key_l = item->valueint;
        printf("key_l = %d\n", key_l);

        item = cJSON_GetObjectItem(root, "time_start"); //拿到节点的time_start
        strcpy(time_start, item->valuestring);
        printf("time_start = %s\n", time_start);

        item = cJSON_GetObjectItem(root, "refresh_l"); //拿到节点的time_start
        refresh_l = item->valueint;
        printf("refresh_l=%d\n",refresh_l);

        item = cJSON_GetObjectItem(root, "time_end"); //拿到节点的time_end
        strcpy(time_end, item->valuestring);
        printf("time_start = %s\n", time_end);
        cJSON_Delete(root);
        putchar('\n');
        MQTTClient_freeMessage(&message);
        MQTTClient_free(topicName);
    }
    return 1;
}

void connlost(void *context, char *cause)
{
    printf("\nConnection lost\n");
    printf("     cause: %s\n", cause);
}
//上报线程
void *publish()
{
    sleep(1); //连接在接收线程，这边先睡一秒让那边建立链接
    //前4个字节存储实际的设备数目
    total = (int *)node_p;
    //判断是哪种模式
    if (type == 0) //0刷新上报，客户端主动采集；
    {
        while (1)
        {
            if (refresh == 1)
            {
                send_qt();
                refresh = 0;
            }
            for (int i = 0; i < *total; i++)
            {
                if (node_arr[i].type == 1)
                {
                    if (node_arr[i].new_val.b_val != node_arr[i].old_val.b_val) //判断新旧值变化
                    {
                        send_qt();
                        break;
                    }
                }
                else if (node_arr[i].type == 2)
                {
                    if (node_arr[i].new_val.i_val != node_arr[i].old_val.i_val) //判断新旧值变化
                    {
                        send_qt();
                        break;
                    }
                }
                else if (node_arr[i].type == 3)
                {
                    if (node_arr[i].new_val.f_val != node_arr[i].old_val.f_val) //判断新旧值变化
                    {
                        send_qt();
                        break;
                    }
                }
            }
            
        }
    }
    else if (type == 1) //1-变化上报，即连续2次值不相等
    {
        while (1)
        {
            for (int i = 0; i < *total; i++)
            {
                if (node_arr[i].type == 1)
                {
                    if (node_arr[i].new_val.b_val != node_arr[i].old_val.b_val) //判断新旧值变化
                    {
                        send_qt();
                        break;
                    }
                }
                else if (node_arr[i].type == 2)
                {
                    if (node_arr[i].new_val.i_val != node_arr[i].old_val.i_val) //判断新旧值变化
                    {
                        send_qt();
                        break;
                    }
                }
                else if (node_arr[i].type == 3)
                {
                    if (node_arr[i].new_val.f_val != node_arr[i].old_val.f_val) //判断新旧值变化
                    {
                        send_qt();
                        break;
                    }
                }
            }
        }
    }
    else if (type == 2) //2-周期上报
    {
        while (1)
        {
            for (int i = 0; i < *total; i++)
            {
                if (node_arr[i].type == 1)
                {
                    if (node_arr[i].new_val.b_val != node_arr[i].old_val.b_val) //判断新旧值变化
                    {
                        send_qt();
                        break;
                    }
                }
                else if (node_arr[i].type == 2)
                {
                    if (node_arr[i].new_val.i_val != node_arr[i].old_val.i_val) //判断新旧值变化
                    {
                        send_qt();
                        break;
                    }
                }
                else if (node_arr[i].type == 3)
                {
                    if (node_arr[i].new_val.f_val != node_arr[i].old_val.f_val) //判断新旧值变化
                    {
                        send_qt();
                        break;
                    }
                }
            }
            send_qt();
            sleep(20);
        }
    }

    MQTTClient_disconnect(client, 10000);
    MQTTClient_destroy(&client);
}
//接收消息
void *subscribe()
{
    int rc;
    int ch;

    MQTTClient_create(&client, ADDRESS, CLIENTID,
                      MQTTCLIENT_PERSISTENCE_NONE, NULL);
    conn_opts.keepAliveInterval = 20;
    conn_opts.cleansession = 1;
    MQTTClient_setCallbacks(client, NULL, connlost, msgarrvd, delivered);
    if ((rc = MQTTClient_connect(client, &conn_opts)) != MQTTCLIENT_SUCCESS)
    {
        printf("Failed to connect, return code %d\n", rc);
        exit(EXIT_FAILURE);
    }
    printf("Subscribing to topic %s\nfor client %s using QoS%d\n\n"
           "Press Q<Enter> to quit\n\n",
           TOPIDR, CLIENTID, QOS);
    MQTTClient_subscribe(client, TOPIDR, QOS); //接受数据
    MQTTClient_subscribe(client, TOPIHR, QOS);//接收历史记录
    do
    {
        ch = getchar();
    } while (ch != 'Q' && ch != 'q');

    MQTTClient_unsubscribe(client, TOPIDR);     //订阅up主题
    MQTTClient_unsubscribe(client, TOPIHR); //订阅history主题
    MQTTClient_disconnect(client, 10000);
    MQTTClient_destroy(&client);
}
//向qt中发送数据
void send_qt()
{
    for (int i = 0; i < (lenth1 + lenth); i++)
    {
        //后面空间存储数据点
        node_arr = (struct std_node *)(node_p + sizeof(int));
        //序列化，将共享内存里的数据进行序列化
        //创建根节点
        cJSON *root = cJSON_CreateObject();
        //创建一个子节点，值是key_k的值
        cJSON *item = cJSON_CreateNumber(node_arr[i].key);
        cJSON_AddItemToObject(root, "key", item);
        //printf("key ==== %d\n",node_arr[i].key);
        item = cJSON_CreateString(node_arr[i].dev_type); //将name放树上
        cJSON_AddItemToObject(root, "name", item);
        //printf("name ==== %s\n",node_arr[i].dev_type);
        item = cJSON_CreateNumber(node_arr[i].type); //将type放树上
        cJSON_AddItemToObject(root, "type", item);
        //printf("type ==== %d\n",node_arr[i].type);
        if (node_arr[i].type == 1) //判断data数据类型，将data放树上
        {
            item = cJSON_CreateBool(node_arr[i].new_val.b_val); //data
            cJSON_AddItemToObject(root, "data", item);
            //printf("%d\n",node_arr[i].new_val.b_val);
        }
        else if (node_arr[i].type == 2)
        {
            item = cJSON_CreateNumber(node_arr[i].new_val.i_val); //data
            //printf("%d\n",node_arr[i].new_val.i_val);
            cJSON_AddItemToObject(root, "data", item);
        }
        else if (node_arr[i].type == 3)
        {
            item = cJSON_CreateNumber(node_arr[i].new_val.f_val); //data
            cJSON_AddItemToObject(root, "data", item);
            //printf("%f\n",node_arr[i].new_val.f_val);
        }
        node_arr[i].old_val = node_arr[i].new_val;//将新值赋值给旧值
        //将json格式发给qt
        char *p = cJSON_PrintUnformatted(root);
        pubmsg.payload = p;
        pubmsg.payloadlen = strlen(p);
        pubmsg.qos = QOS;
        pubmsg.retained = 0;
        MQTTClient_publishMessage(client,"data_send", &pubmsg, &token);
        // printf("Waiting for up to %d seconds for publication of down"
        //        "on topic %s for client with ClientID: %s\n",
        //        (int)(TIMEOUT / 1000),TOPIDR, CLIENTID);
        
        cJSON_Delete(root);
    }
    printf("上报一次数据\n");
}
//给qt发送的历史记录线程
void *publish_h()
{
    sleep(1);
    while (1)
    {
        if (refresh_l == 1)
        {
            char sql[128] = {0}; //sq
            char **result = NULL;
            int row = 0;    // 存放有几行符合条件的信息
            int column = 0; // 列数SELECT * FROM data ORDER BY time DESC LIMIT 10;
            //SELECT * FROM my_table WHERE datetime(event_time) BETWEEN '2022-01-01 00:00:00' AND '2022-12-31 23:59:59';

            sprintf(sql, "select * from history where key = %d and time BETWEEN \"%s\" AND \"%s\";", key_l,time_start,time_end);

            // sprintf(sql, "select * from history where key = %d order by time desc limit 10;", key_l);
            sqlite3_get_table(db, sql, &result, &row, &column, &errmsg);
            //创建根节点
            cJSON *root = cJSON_CreateObject();
            cJSON *data = cJSON_CreateArray();
            cJSON *item;
            
            printf("row == %d\n",row);
            int k = 0;
            for (int i = 0; i <= row; i++)
            {
                for (int j = 0; j < column; j++)
                    printf("%s ", result[k++]);
                putchar(10);
            }
            int t = 3;
            for (int i = 0; i < row; i++)
            {
                cJSON *dev = cJSON_CreateObject();
                item = cJSON_CreateString(result[t++]); //把key值放到dev节点上
                cJSON_AddItemToObject(dev, "key", item);
                item = cJSON_CreateString(result[t++]); //把time值放到dev节点上
                cJSON_AddItemToObject(dev, "time", item);
                item = cJSON_CreateString(result[t++]); //把data值放到dev节点上
                cJSON_AddItemToObject(dev, "data", item);

                cJSON_AddItemToArray(data, dev); //将dev节点放到数组节点上
            }
            cJSON_AddItemToObject(root,"history", data);
            //将json格式发给qt
            char *p = cJSON_PrintUnformatted(root);
            pubmsg.payload = p;
            pubmsg.payloadlen = strlen(p);
            pubmsg.qos = QOS;
            pubmsg.retained = 0;
            MQTTClient_publishMessage(client, "history_send", &pubmsg, &token);
            // printf("Waiting for up to %d seconds for publication of down"
            //        "on topic %s for client with ClientID: %s\n",
            //        (int)(TIMEOUT / 1000), "history", CLIENTID);
            cJSON_Delete(root);
            refresh_l = 0;
        }
    }
}
//数据库线程
void *sqlite()
{
    sleep(1);
    // 打开数据库
    if (sqlite3_open("./history.db", &db) != SQLITE_OK)
    {
        fprintf(stderr, "open err:%s\n", sqlite3_errmsg(db));
        return NULL;
    }
    printf("open ok\n");

    // 创建一张表
    if (sqlite3_exec(db, "create table history(key int,time char,value float);",
                     NULL, NULL, &errmsg) != SQLITE_OK)
    {
        fprintf(stderr, "create err:%s\n", errmsg);
        // return -1;
    }
    char sql[128] = {0}; //sq
    while (1)
    {
        sleep(20);
        char date[32] = "";
        time_t t;
        struct tm *tp;
        time(&t);
        tp = localtime(&t);
        /*
        sprintf(date, "%d-%02d-%02d %02d:%02d:%02d", tp->tm_year+1900,
            tp->tm_mon+1, tp->tm_mday, tp->tm_hour, tp->tm_min, tp->tm_sec);*/
        strftime(date, 64, "%Y-%m-%d %H:%M:%S", tp); //将时间字符串放到data里面
        printf("%s\n", date);
        for (int i = 0; i < (lenth + lenth1); i++)
        {
            if (node_arr[i].type == 1)
            {
                sprintf(sql, "insert into history values(%d,\"%s\",%f);", node_arr[i].key, date, node_arr[i].new_val.b_val);
                //z执行sql语句插入表
                if (sqlite3_exec(db, sql, NULL, NULL, &errmsg) != SQLITE_OK)
                {
                    fprintf(stderr, "insert err:%s\n", errmsg);
                }
                //printf("insert ok\n");
            }
            else if (node_arr[i].type == 2)
            {
                sprintf(sql, "insert into history values(%d,\"%s\",%f);", node_arr[i].key, date, node_arr[i].new_val.i_val);
                //z执行sql语句插入表
                if (sqlite3_exec(db, sql, NULL, NULL, &errmsg) != SQLITE_OK)
                {
                    fprintf(stderr, "insert err:%s\n", errmsg);
                }
                //printf("insert ok\n");
            }
            else if (node_arr[i].type == 3)
            {
                sprintf(sql, "insert into history values(%d,\"%s\",%f);", node_arr[i].key, date, node_arr[i].new_val.f_val);
                //z执行sql语句插入表
                if (sqlite3_exec(db, sql, NULL, NULL, &errmsg) != SQLITE_OK)
                {
                    fprintf(stderr, "insert err:%s\n", errmsg);
                }
                //printf("insert ok\n");
            }
            memset(sql, 0, sizeof(sql));
        }
        printf("插入一次数据\n");
    }
}